package db

import (
	"context"
	"fmt"
	"log/slog"
	"net/url"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

var (
	RabbitMQClients      = make(map[string]*RabbitMQClient)
	rabbitMQClientsMutex sync.Mutex
)

// RabbitMQClient 是 RabbitMQ 客户端结构体
type RabbitMQClient struct {
	conn         *amqp.Connection // 连接
	channel      *amqp.Channel    // 通道
	amqpURL      string           // AMQP URL
	exchange     string           // 交换机名称
	exchangeType string           // 交换机类型
	routingKey   string           // 路由键
	queue        string           // 队列名称
	args         amqp.Table       // 额外参数
	closeChannel chan *amqp.Error // 通道关闭事件
	mutex        sync.Mutex       // 互斥锁
}

// NewClient 创建一个新的 RabbitMQClient 实例
func NewClient(amqpURL, exchange, exchangeType, routingKey, queue string, args amqp.Table) (*RabbitMQClient, error) {
	client := &RabbitMQClient{
		amqpURL:      amqpURL,
		exchange:     exchange,
		exchangeType: exchangeType,
		routingKey:   routingKey,
		queue:        queue,
		args:         args,
	}

	// 调用maskHalfPasswordURL函数隐藏URL中密码的后半部分
	hiddenURL, _ := MaskHalfPasswordURL(amqpURL, '-')

	err := client.connect()
	if err != nil {
		slog.Error("RabbitMQ客户端连接失败",
			slog.String("url", hiddenURL),
			slog.String("exchange", exchange),
			slog.String("exchange_type", exchangeType),
			slog.String("routing_key", routingKey),
			slog.String("queue", queue))
		return nil, err
	}

	parsedURL, err := url.Parse(amqpURL)
	if err != nil {
		return nil, err
	}
	vhost := parsedURL.Path

	slog.Info("RabbitMQ客户端连接成功",
		slog.String("url", hiddenURL),
		slog.String("exchange", exchange),
		slog.String("exchange_type", exchangeType),
		slog.String("routing_key", routingKey),
		slog.String("queue", queue),
		slog.String("vhost", vhost))

	return client, nil
}

// FetchOrCreateClient 获取或创建一个 RabbitMQClient 实例
func FetchOrCreateClient(amqpURL, exchange, exchangeType, routingKey, queue string, args amqp.Table) (*RabbitMQClient, error) {
	rabbitMQClientsMutex.Lock()
	defer rabbitMQClientsMutex.Unlock()

	parsedURL, err := url.Parse(amqpURL)
	if err != nil {
		return nil, err
	}
	vhost := parsedURL.Path

	key := fmt.Sprintf("%s_%s_%s_%s_%s", vhost, exchange, exchangeType, routingKey, queue)
	if client, ok := RabbitMQClients[key]; ok {
		return client, nil
	}

	client, err := NewClient(amqpURL, exchange, exchangeType, routingKey, queue, args)
	if err != nil {
		return nil, err
	}

	RabbitMQClients[key] = client

	return client, nil
}

// connect 连接到 RabbitMQ 服务器并创建通道
func (c *RabbitMQClient) connect() error {
	var err error

	// 创建自定义配置并设置心跳
	config := amqp.Config{
		Heartbeat: 5 * time.Second, // 设置为5秒, 默认为10秒
		Locale:    "zh_CN",
	}

	// 使用 DialConfig 和自定义配置替换 amqp.Dial
	c.conn, err = amqp.DialConfig(c.amqpURL, config)
	if err != nil {
		return err
	}

	c.channel, err = c.conn.Channel()
	if err != nil {
		return err
	}

	// 创建交换机
	err = c.channel.ExchangeDeclare(
		c.exchange,     // 交换机名称
		c.exchangeType, // 交换机类型
		true,           // 是否持久化
		false,          // 是否自动删除
		false,          // 是否内部
		false,          // 是否无需等待
		nil,            // 额外参数
	)
	if err != nil {
		return err
	}

	if c.queue != "" && c.routingKey != "" {
		// 创建队列
		_, err = c.channel.QueueDeclare(
			c.queue, // 队列名称
			true,    // 是否持久化
			false,   // 是否自动删除
			false,   // 是否排他
			false,   // 是否无需等待
			c.args,
		)
		if err != nil {
			return err
		}

		// 绑定队列到交换机
		err = c.channel.QueueBind(
			c.queue,      // 队列名称
			c.routingKey, // 路由键
			c.exchange,   // 交换机名称
			false,        // 是否无需等待, 这里设为 false
			nil,          // 附加参数, 这里设为 nil
		)
		if err != nil {
			return err
		}
	}

	// 注册 NotifyClose 监听器
	c.closeChannel = c.channel.NotifyClose(make(chan *amqp.Error))

	// 在单独的 goroutine 中处理断开连接事件
	go c.handleReconnect()

	return nil
}

// handleReconnect 处理 RabbitMQ 客户端断开连接的情况
func (c *RabbitMQClient) handleReconnect() {
	closeErr := <-c.closeChannel
	slog.Error("RabbitMQ通道或连接关闭", slog.String("error", closeErr.Error()))

	// 从RabbitMQClients中删除当前客户端
	c.removeFromClients()

	// 重连逻辑
	backoffTime := 1 * time.Second
	maxBackoffTime := 30 * time.Second
	maxRetries := 10
	for retries := 0; retries < maxRetries; retries++ {
		slog.Info("尝试重新连接RabbitMQ")
		err := c.connect()
		if err == nil {
			slog.Info("成功重新连接RabbitMQ")
			// 重新添加到RabbitMQClients
			c.addToClients()
			return
		}

		if backoffTime < maxBackoffTime {
			backoffTime *= 2
			if backoffTime > maxBackoffTime {
				backoffTime = maxBackoffTime
			}
		}

		slog.Error(fmt.Sprintf("无法重新连接到RabbitMQ, 等待%v后重试", backoffTime), slog.String("url", c.amqpURL))
		time.Sleep(backoffTime)
	}
	slog.Error("重试次数已达到最大值，无法重新连接RabbitMQ", slog.String("url", c.amqpURL))
}

// removeFromClients 从RabbitMQClients中删除当前客户端
func (c *RabbitMQClient) removeFromClients() {
	rabbitMQClientsMutex.Lock()
	defer rabbitMQClientsMutex.Unlock()

	parsedURL, err := url.Parse(c.amqpURL)
	if err != nil {
		slog.Error("解析URL失败", slog.String("error", err.Error()))
		return
	}
	vhost := parsedURL.Path

	key := fmt.Sprintf("%s_%s_%s_%s_%s", vhost, c.exchange, c.exchangeType, c.routingKey, c.queue)
	delete(RabbitMQClients, key)
}

// addToClients 将当前客户端添加到RabbitMQClients
func (c *RabbitMQClient) addToClients() {
	rabbitMQClientsMutex.Lock()
	defer rabbitMQClientsMutex.Unlock()

	parsedURL, err := url.Parse(c.amqpURL)
	if err != nil {
		slog.Error("解析URL失败", slog.String("error", err.Error()))
		return
	}
	vhost := parsedURL.Path

	key := fmt.Sprintf("%s_%s_%s_%s_%s", vhost, c.exchange, c.exchangeType, c.routingKey, c.queue)
	RabbitMQClients[key] = c
}

// Close 关闭 RabbitMQ 客户端连接和通道
func (c *RabbitMQClient) Close() {
	c.removeFromClients()
	if c.channel != nil {
		c.channel.Close()
	}
	if c.conn != nil {
		c.conn.Close()
	}
}

// Publish 向指定的交换机和路由键发送消息
func (c *RabbitMQClient) Publish(payload amqp.Publishing) error {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	err := c.channel.PublishWithContext(
		context.Background(),
		c.exchange,   // 交换机名称
		c.routingKey, // 路由键
		false,        // mandatory, 是否要求消息必须被路由到队列（即至少有一个队列绑定到此路由键）, 否则将返回给生产者
		false,        // immediate, 是否要求消息必须立即被消费者接收, 否则将返回给生产者
		payload,      // 载荷, 内含消息及其属性
	)
	if err != nil {
		return err
	}

	return nil
}

// MaskHalfPasswordURL 掩藏 URL 中的密码
func MaskHalfPasswordURL(urlStr string, maskChar rune) (string, error) {
	// 解析URL
	u, err := url.Parse(urlStr)
	if err != nil {
		return "", fmt.Errorf("error parsing URL: %v", err)
	}

	// 获取用户名和密码
	username := u.User.Username()
	password, _ := u.User.Password()

	// 计算需要掩藏的字符数量
	length := len(password)
	numMask := length / 2

	// 替换密码后半部分
	maskedPassword := make([]rune, length)
	for i, char := range password {
		if i >= numMask {
			maskedPassword[i] = maskChar
		} else {
			maskedPassword[i] = char
		}
	}

	// 创建新的用户信息
	newUserInfo := url.UserPassword(username, string(maskedPassword))

	// 更新URL
	u.User = newUserInfo

	// 返回更新后的URL字符串
	return u.String(), nil
}
